public class PrefNeighbourCalThread extends Thread {

	private SharedMessageQueue[] sharedMsg;

	private boolean[] UnchokedNeighbors; // TODO: replace with N; check if

	// initialized to false by default

	private int optimisticallyUnchokedNeighbor;// peerID of OUN

	private PreferredNeighborsSelector PN;

	private OptimisticallyUnchokedSelector OUN;

	private int unchokingInterval, optUnchokeInterval;

	private int k; // numberOfPrefferedNeighbors

	private int numOfPeers;

	private Object lock;

	private Logger logger;
	int peerId;

	PrefNeighbourCalThread(PeerData peerData, SharedMessageQueue[] sharedMsg,
			Logger logger) {

		this.sharedMsg = sharedMsg;

		this.unchokingInterval = peerData.getUnchokingInterval();

		this.optUnchokeInterval = peerData.getOptimisticUnchokingInterval();

		k = peerData.getNumberOfPreferredNeighbors();

		this.numOfPeers = peerData.getNumOfPeers();

		this.UnchokedNeighbors = new boolean[peerData.getNumOfPeers()];

		this.logger = logger;
		this.peerId = peerData.getPeerID();

		lock = new Object();

		PN = new PreferredNeighborsSelector(peerData, "PN");

		OUN = new OptimisticallyUnchokedSelector(peerData, "OUN");

	}

	public void run() {

		PN.start();

		OUN.start();



		try {

			PN.join();

			OUN.join();

			 logger.log("Peer " + peerId + " has downloaded the complete file. ");
			logger.endLog();
			
			
			System.err.println("terminate "+peerId);

			System.out.println("terminate "+peerId);

		} catch (InterruptedException e) {

			e.printStackTrace();

		}

	}

	class PreferredNeighborsSelector extends Thread {

		PeerData peerData;

		Message chokeMessage;

		Message unchokeMessage;

		int i = 0;
		int prefNghbrNum;

		int[] unchokeTheseNeighbors = new int[5]; // TODO: replace 5 with k;

		// null if no one

		// interested; -1 indicates

		// <k people interested

		// contains peerNumbers

		PreferredNeighborsSelector(PeerData peerData, String name) {

			super(name);

			this.peerData = peerData;

			chokeMessage = new ChokeMessage();

			unchokeMessage = new UnchokeMessage();
			prefNghbrNum = k;

		}

		public void run() {

			while (!peerData.checkForTermination()) {
				prefNghbrNum = k;
				synchronized (lock) {

					if (optimisticallyUnchokedNeighbor != -1)
						prefNghbrNum++;

					// Select K nghbrs with highest download rate from I

					unchokeTheseNeighbors = peerData
							.topKNeighbors(prefNghbrNum);

					// Reset rates to 0 except for OUN

					peerData.resetDownloadExcept(optimisticallyUnchokedNeighbor);

					// Send unchoke messages

					for (int peerNum = 0; peerNum < prefNghbrNum; peerNum++) {

						if ((containsInArray(unchokeTheseNeighbors, peerNum)) && (unchokeTheseNeighbors[peerNum] != -1)

								&& (!UnchokedNeighbors[unchokeTheseNeighbors[peerNum]])
								&& peerNum != peerData.getPeerNum(peerData
										.getPeerID())) {

							// TODO: check with sirisha

							System.out.println("peer: " + peerData.getPeerID()
									+ " sending unchoke");

							sharedMsg[peerNum].enQueue(unchokeMessage);

							synchronized (sharedMsg[peerNum].dummyObject) {

								sharedMsg[peerNum].dummyObject.notify();

							}

							UnchokedNeighbors[peerNum] = true;

						}

					}

					// Send choke messages

					for (int peerNum = 0; peerNum < numOfPeers; peerNum++) {

						if ((UnchokedNeighbors[peerNum])

								&& (chokeInNextRound(peerNum))
								&& peerNum != peerData.getPeerNum(peerData
										.getPeerID())) {

							// TODO: check with sirisha

							System.out.println("peer: " + peerData.getPeerID()
									+ " sending choke");

							sharedMsg[peerNum].enQueue(chokeMessage);

							synchronized (sharedMsg[peerNum].dummyObject) {

								sharedMsg[peerNum].dummyObject.notify();

							}

							UnchokedNeighbors[peerNum] = false;

						}

					}

					String prefNghbrStr = toString(UnchokedNeighbors);

					if (!prefNghbrStr.equals(" "))

						logger.log("Peer " + peerData.getPeerID()

						+ " has the preffered neighbors "

						+ prefNghbrStr + ".");

				} // Release lock

				try {

					Thread.sleep(unchokingInterval * 1000);

				} catch (InterruptedException e) {

					System.out.println("PrefNeighborCalThread Interrupted!!!!");

					e.printStackTrace();

				}

			}

		}

		boolean chokeInNextRound(int peerNumber) {

			for (int i = 0; i < prefNghbrNum; i++) {

				if (unchokeTheseNeighbors[i] == peerNumber)

					return false;

			}

			return true;

		}

		private String toString(boolean[] Neighbors) {

			String str = " ";

			boolean firstPrefNghbr = true; // dont prepend ',' to string if so

			for (int i = 0; i < Neighbors.length; i++) {

				if (Neighbors[i] && i != optimisticallyUnchokedNeighbor) {

					if (!firstPrefNghbr)

						str = str + " , ";

					str = str + peerData.getPeerID(i);

					firstPrefNghbr = false;

				}

			}

			return str;

		}
      
       // helper function

     private boolean containsInArray(int array[], int checkMe){

             for(int i=0; i<array.length; i++){

                     if(array[i] == checkMe)

                             return true;

             }             

             return false;

     }

	}

	class OptimisticallyUnchokedSelector extends Thread {

		PeerData peerData;

		int unchokeMe;

		Message unchokeMessage;

		public OptimisticallyUnchokedSelector(PeerData peerData, String name) {

			super(name);

			this.peerData = peerData;

			unchokeMessage = new UnchokeMessage();

		}

		public void run() {

			while (!peerData.checkForTermination()) {

				// Select one optimistically UN from I-UN-OUN

				synchronized (lock) {

					unchokeMe = peerData

					.nextOptUnchokedNeighbor(UnchokedNeighbors);

					if (unchokeMe != -1)

						UnchokedNeighbors[unchokeMe] = true;

					optimisticallyUnchokedNeighbor = unchokeMe;

				}

				if (unchokeMe != -1) {

					// send unchoke message

					// TODO: check with sirisha

					System.out.println("peer: " + peerData.getPeerID()
							+ " sending OptUnchoke");

					sharedMsg[unchokeMe].enQueue(unchokeMessage);

					synchronized (sharedMsg[unchokeMe].dummyObject) {

						sharedMsg[unchokeMe].dummyObject.notify();

					}

					logger.log("Peer " + peerData.getPeerID()

					+ " has the optimistically unchoked neighbor "

					+ peerData.getPeerID(unchokeMe) + ".");

				}

				try {

					Thread.sleep(optUnchokeInterval * 1000);

				} catch (InterruptedException e) {

					System.out

					.println("optUnchokeNeighborCalThread Interrupted!!!!");

					e.printStackTrace();

				}

			}

		}

	}

}